package com.wt.springsamples.kafka.demo;

import com.wt.springsamples.kafka.Constants;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 *需要配置broker的listeners 、 advertised.listeners参数值，默认被注释
 * 如：listeners=PLAINTEXT://0.0.0.0:9092 、advertised.listeners=PLAINTEXT://172.22.240.20:9092
 */
public class KafkaProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,Constants.CLUSTER_SERVER);
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.RETRIES_CONFIG,3);//重试次数
        properties.put("acks","1");//0-不等待broker确认，1-等待leader确认，all-等待所有节点确认
        //同步发送，

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(properties)) {
            int i = 1;
            while (true) {
                TimeUnit.SECONDS.sleep(1);
                ProducerRecord<String, String> record = new ProducerRecord<>(Constants.TOPIC_NAME, "key" + i, "value" + i);
                Future<RecordMetadata> future = producer.send(record);
                RecordMetadata recordMetadata = future.get();
                System.out.println(record.value()+"同步发送成功，大小:" + recordMetadata.serializedValueSize() + ",主题:" + recordMetadata.topic() + "，分区:" + recordMetadata.partition());
                i++;
                //异步发送
//                ProducerRecord<String, String> record2 = new ProducerRecord<>(Constants.TOPIC_NAME, "async key" + i, "async value" + i);
//                producer.send(record2, (metadata, exception) -> {
//                    if (exception != null) {
//                        exception.printStackTrace();
//                        //todo 发送消息失败，需要处理异常,可以将发送失败的数据入库或开启重试机制
//                    } else {
//                        System.out.println("异步发送成功，大小:" + metadata.serializedValueSize() + ",主题:" + metadata.topic() + "，分区:" + metadata.partition());
//                    }
//                });
            }

        } catch (Exception exception) {
            exception.printStackTrace();
            throw new RuntimeException("error");
        }
    }
}
